{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.core.interactiveshell import InteractiveShell\n",
    "InteractiveShell.ast_node_interactivity = 'all'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.启动与停止线程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "T-minusThread Still running\n",
      "Thread waiting finish\n",
      " 4\n",
      "T-minus 3\n",
      "T-minus 2\n",
      "T-minus 1\n",
      "Thread finished\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "def countdown(n):\n",
    "    while n > 0:\n",
    "        print('T-minus', n)\n",
    "        n -= 1\n",
    "        time.sleep(1)\n",
    "\n",
    "from threading import Thread\n",
    "\n",
    "# 创建线程\n",
    "t = Thread(target=countdown, args=(4,))\n",
    "# 启动线程\n",
    "t.start()\n",
    "\n",
    "# 查看线程状态\n",
    "if t.is_alive():\n",
    "    print('Thread Still running')\n",
    "else:\n",
    "    print('Thread Completed')\n",
    "\n",
    "# 将线程加入到当前线程，并等待它终止\n",
    "print('Thread waiting finish')\n",
    "t.join()\n",
    "print('Thread finished')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.判断线程是否已经启动"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Launching countdown\n",
      "countdown starting\n",
      "T-minus 4\n",
      "countdown is running\n",
      "Thread waiting finish\n",
      "T-minus 3\n",
      "T-minus 2\n",
      "T-minus 1\n",
      "Thread finished\n"
     ]
    }
   ],
   "source": [
    "from threading import Thread, Event\n",
    "import time\n",
    "\n",
    "def countdown(n, started_evt):\n",
    "    print('countdown starting')\n",
    "    started_evt.set()\n",
    "    while n > 0:\n",
    "        print('T-minus', n)\n",
    "        n -= 1\n",
    "        time.sleep(1)\n",
    "\n",
    "# Create the event object that will be used to signal startup\n",
    "# event 对象最好单次使用，一旦这个对象被设置为真，你就应该丢弃它\n",
    "started_evt = Event()\n",
    "\n",
    "# Launch the thread and pass the startup event\n",
    "print('Launching countdown')\n",
    "t = Thread(target=countdown, args=(4,started_evt))\n",
    "t.start()\n",
    "\n",
    "\n",
    "# Wait for the thread to start\n",
    "started_evt.wait()\n",
    "print('countdown is running')\n",
    "\n",
    "# 将线程加入到当前线程，并等待它终止\n",
    "print('Thread waiting finish')\n",
    "t.join()\n",
    "print('Thread finished')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "如果一个线程需要不停地重复使用 event 对象，你最好使用 Condition 对象来代替。\n",
    "\n",
    "如果你只想唤醒单个线程，最好是使用信号量或者 Condition 对象来替代。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.线程间通信\n",
    "\n",
    "从一个线程向另一个线程发送数据最安全的方式可能就是使用 queue 库中的队列了。创建一个被多个线程共享的 Queue 对象，这些线程通过使用 put() 和 get() 操作来向队列中添加或者删除元素。 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from queue import Queue\n",
    "from threading import Thread\n",
    "\n",
    "# A thread that produces data\n",
    "def producer(out_q):\n",
    "    while True:\n",
    "        # Produce some data\n",
    "        ...\n",
    "        out_q.put(data)\n",
    "\n",
    "# A thread that consumes data\n",
    "def consumer(in_q):\n",
    "    while True:\n",
    "# Get some data\n",
    "        data = in_q.get()\n",
    "        # Process the data\n",
    "        ...\n",
    "\n",
    "# Create the shared queue and launch both threads\n",
    "q = Queue()\n",
    "t1 = Thread(target=consumer, args=(q,))\n",
    "t2 = Thread(target=producer, args=(q,))\n",
    "t1.start()\n",
    "t2.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.给关键部分加锁\n",
    "\n",
    "要在多线程程序中安全使用可变对象，你需要使用 threading 库中的 Lock 对象。\n",
    "\n",
    "```python\n",
    "import threading\n",
    "lock = threading.Lock()\n",
    "lock.acquire()\n",
    "pass\n",
    "lock.release()\n",
    "```\n",
    "\n",
    "Lock 对象和 with 语句块一起使用可以保证互斥执行，就是每次只有一个线程可以执行 with 语句包含的代码块。\n",
    "\n",
    "with 语句会在这个代码块执行前自动获取锁，在执行结束后自动释放锁。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "\n",
    "class SharedCounter:\n",
    "    def __init__(self, init_value=0):\n",
    "        self._value = init_value\n",
    "        self._value_lock = threading.Lock()\n",
    "\n",
    "    def incr(self, delta=1):\n",
    "        with self._value_lock:\n",
    "            self._value += delta\n",
    "    \n",
    "    def decr(self, delta=1):\n",
    "        with self._value_lock:\n",
    "            self._value -= delta"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.防止死锁的加锁机制\n",
    "\n",
    "在多线程程序中，死锁问题很大一部分是由于线程同时获取多个锁造成的。\n",
    "\n",
    "解决死锁问题的一种方案是为程序中的每一个锁分配一个唯一的id，然后只允许按照升序规则来使用多个锁。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6.保存线程的状态信息\n",
    "\n",
    "有时在多线程编程中，你需要只保存当前运行线程的状态。 \n",
    "\n",
    "要这么做，可使用 thread.local() 创建一个本地线程存储对象。 对这个对象的属性的保存和读取操作都只会对执行线程可见，而其他线程并不可见。\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from socket import socket, AF_INET, SOCK_STREAM\n",
    "import threading\n",
    "\n",
    "# 代码中，自己观察对于 self.local 属性的使用。 \n",
    "# 它被初始化为一个 threading.local() 实例。 \n",
    "# 其他方法操作被存储为 self.local.sock 的套接字对象。 \n",
    "# 有了这些就可以在多线程中安全的使用 LazyConnection 实例了。\n",
    "class LazyConnection:\n",
    "    def __init__(self, address, family=AF_INET, type=SOCK_STREAM):\n",
    "        self.address = address\n",
    "        self.family = AF_INET\n",
    "        self.type = SOCK_STREAM\n",
    "        self.local = threading.local()\n",
    "\n",
    "    def __enter__(self):\n",
    "        if hasattr(self.local, 'sock'):\n",
    "            raise RuntimeError('Already connected')\n",
    "        self.local.sock = socket(self.family, self.type)\n",
    "        self.local.sock.connect(self.address)\n",
    "        return self.local.sock\n",
    "\n",
    "    def __exit__(self, exc_ty, exc_val, tb):\n",
    "        self.local.sock.close()\n",
    "        del self.local.sock\n",
    "\n",
    "\n",
    "from functools import partial\n",
    "def test(conn):\n",
    "    with conn as s:\n",
    "        s.send(b'GET /index.html HTTP/1.0\\r\\n')\n",
    "        s.send(b'Host: www.python.org\\r\\n')\n",
    "\n",
    "        s.send(b'\\r\\n')\n",
    "        resp = b''.join(iter(partial(s.recv, 8192), b''))\n",
    "\n",
    "    print('Got {} bytes'.format(len(resp)))\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    conn = LazyConnection(('www.python.org', 80))\n",
    "\n",
    "    # 它之所以行得通的原因是每个线程会创建一个自己专属的套接字连接（存储为self.local.sock）。 \n",
    "    # 因此，当不同的线程执行套接字操作时，由于操作的是不同的套接字，因此它们不会相互影响。\n",
    "    t1 = threading.Thread(target=test, args=(conn,))\n",
    "    t2 = threading.Thread(target=test, args=(conn,))\n",
    "    t1.start()\n",
    "    t2.start()\n",
    "    t1.join()\n",
    "    t2.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 7.创建一个线程池\n",
    "\n",
    "concurrent.futures 函数库有一个 ThreadPoolExecutor 类可以被用来完成这个任务。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from socket import AF_INET, SOCK_STREAM, socket\n",
    "from concurrent.futures import ThreadPoolExecutor\n",
    "\n",
    "def echo_client(sock, client_addr):\n",
    "    '''\n",
    "    Handle a client connection\n",
    "    '''\n",
    "    print('Got connection from', client_addr)\n",
    "    while True:\n",
    "        msg = sock.recv(65536)\n",
    "        if not msg:\n",
    "            break\n",
    "        sock.sendall(msg)\n",
    "    print('Client closed connection')\n",
    "    sock.close()\n",
    "\n",
    "def echo_server(addr):\n",
    "    pool = ThreadPoolExecutor(128)\n",
    "    sock = socket(AF_INET, SOCK_STREAM)\n",
    "    sock.bind(addr)\n",
    "    sock.listen(5)\n",
    "    while True:\n",
    "        client_sock, client_addr = sock.accept()\n",
    "        pool.submit(echo_client, client_sock, client_addr)\n",
    "\n",
    "echo_server(('',15000))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 8.简单的并行编程\n",
    "\n",
    "concurrent.futures 库提供了一个 ProcessPoolExecutor 类， 可被用来在一个单独的Python解释器中执行计算密集型函数。\n",
    "\n",
    "ProcessPoolExecutor 的典型用法如下：\n",
    "\n",
    "```python\n",
    "from concurrent.futures import ProcessPoolExecutor\n",
    "\n",
    "with ProcessPoolExecutor() as pool:\n",
    "    ...\n",
    "    do work in parallel using pool\n",
    "    ...\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "假定你有个Apache web服务器日志目录的gzip压缩包：\n",
    "\n",
    "```\n",
    "logs/\n",
    "   20120701.log.gz\n",
    "   20120702.log.gz\n",
    "   20120703.log.gz\n",
    "   20120704.log.gz\n",
    "   20120705.log.gz\n",
    "   20120706.log.gz\n",
    "   ...\n",
    "```\n",
    "\n",
    "进一步假设每个日志文件内容类似下面这样：\n",
    "\n",
    "```\n",
    "124.115.6.12 - - [10/Jul/2012:00:18:50 -0500] \"GET /robots.txt ...\" 200 71\n",
    "210.212.209.67 - - [10/Jul/2012:00:18:51 -0500] \"GET /ply/ ...\" 200 11875\n",
    "210.212.209.67 - - [10/Jul/2012:00:18:51 -0500] \"GET /favicon.ico ...\" 404 369\n",
    "61.135.216.105 - - [10/Jul/2012:00:20:04 -0500] \"GET /blog/atom.xml ...\" 304 -\n",
    "...\n",
    "\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 下面是一个脚本，在这些日志文件中查找出所有访问过robots.txt文件的主机\n",
    "# findrobots.py\n",
    "\n",
    "import gzip\n",
    "import io\n",
    "import glob\n",
    "\n",
    "def find_robots(filename):\n",
    "    '''\n",
    "    Find all of the hosts that access robots.txt in a single log file\n",
    "    '''\n",
    "    robots = set()\n",
    "    with gzip.open(filename) as f:\n",
    "        for line in io.TextIOWrapper(f,encoding='ascii'):\n",
    "            fields = line.split()\n",
    "            if fields[6] == '/robots.txt':\n",
    "                robots.add(fields[0])\n",
    "    return robots\n",
    "\n",
    "# 使用单核CPU\n",
    "def find_all_robots(logdir):\n",
    "    '''\n",
    "    Find all hosts across and entire sequence of files\n",
    "    '''\n",
    "    files = glob.glob(logdir+'/*.log.gz')\n",
    "    all_robots = set()\n",
    "    # 直接使用map，使用单核CPU\n",
    "    for robots in map(find_robots, files):\n",
    "        all_robots.update(robots)\n",
    "    return all_robots\n",
    "\n",
    "\n",
    "# 使用多核CPU\n",
    "from concurrent import futures\n",
    "def find_all_robots_ncpu(logdir):\n",
    "    '''\n",
    "    Find all hosts across and entire sequence of files\n",
    "    '''\n",
    "    files = glob.glob(logdir+'/*.log.gz')\n",
    "    all_robots = set()\n",
    "    # 使用ProcessPoolExecutor，使用多核CPU\n",
    "    with futures.ProcessPoolExecutor() as pool:\n",
    "        for robots in pool.map(find_robots, files):\n",
    "            all_robots.update(robots)\n",
    "    return all_robots\n",
    "\n",
    "# 运行\n",
    "if __name__ == '__main__':\n",
    "    robots = find_all_robots_ncpu('logs')\n",
    "    for ipaddr in robots:\n",
    "        print(ipaddr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 9.Python的全局锁问题\n",
    "\n",
    "主要是讲了全局解释器锁GIL的相关理论。\n",
    "\n",
    "跳过。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 10.定义一个Actor任务"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Got: Hello\n",
      "Got: World\n"
     ]
    }
   ],
   "source": [
    "from queue import Queue\n",
    "from threading import Thread, Event\n",
    "\n",
    "# Sentinel used for shutdown\n",
    "class ActorExit(Exception):\n",
    "    pass\n",
    "\n",
    "class Actor:\n",
    "    def __init__(self):\n",
    "        self._mailbox = Queue()\n",
    "\n",
    "    def send(self, msg):\n",
    "        '''\n",
    "        Send a message to the actor\n",
    "        '''\n",
    "        self._mailbox.put(msg)\n",
    "\n",
    "    def recv(self):\n",
    "        '''\n",
    "        Receive an incoming message\n",
    "        '''\n",
    "        msg = self._mailbox.get()\n",
    "        if msg is ActorExit:\n",
    "            raise ActorExit()\n",
    "        return msg\n",
    "\n",
    "    def close(self):\n",
    "        '''\n",
    "        Close the actor, thus shutting it down\n",
    "        '''\n",
    "        self.send(ActorExit)\n",
    "\n",
    "    def start(self):\n",
    "        '''\n",
    "        Start concurrent execution\n",
    "        '''\n",
    "        self._terminated = Event()\n",
    "        t = Thread(target=self._bootstrap)\n",
    "\n",
    "        t.daemon = True\n",
    "        t.start()\n",
    "\n",
    "    def _bootstrap(self):\n",
    "        try:\n",
    "            self.run()\n",
    "        except ActorExit:\n",
    "            pass\n",
    "        finally:\n",
    "            self._terminated.set()\n",
    "\n",
    "    def join(self):\n",
    "        self._terminated.wait()\n",
    "\n",
    "    def run(self):\n",
    "        '''\n",
    "        Run method to be implemented by the user\n",
    "        '''\n",
    "        while True:\n",
    "            msg = self.recv()\n",
    "\n",
    "# Sample ActorTask\n",
    "class PrintActor(Actor):\n",
    "    def run(self):\n",
    "        while True:\n",
    "            msg = self.recv()\n",
    "            print('Got:', msg)\n",
    "\n",
    "# Sample use\n",
    "p = PrintActor()\n",
    "p.start()\n",
    "p.send('Hello')\n",
    "p.send('World')\n",
    "p.close()\n",
    "p.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 11.实现消息发布/订阅模型\n",
    "\n",
    "跳过。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 12.使用生成器代替线程\n",
    "\n",
    "使用生成器（协程）替代系统线程来实现并发，这个有时又被称为用户级线程或绿色线程。\n",
    "\n",
    "https://python3-cookbook.readthedocs.io/zh_CN/latest/c12/p12_using_generators_as_alternative_to_threads.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 13.多个线程队列轮询\n",
    "\n",
    "没看懂。\n",
    "\n",
    "https://python3-cookbook.readthedocs.io/zh_CN/latest/c12/p13_polling_multiple_thread_queues.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 14.在Unix系统上面启动守护进程\n",
    "\n",
    "可以用py写一个程序，然后用linux的systemd系统来把这个py程序设置为服务并且支持开机启动。\n",
    "\n",
    "非要用写代码的方式做成守护进程，真的有些复杂了。\n",
    "\n",
    "跳过，用不到这个。\n",
    "\n",
    "https://python3-cookbook.readthedocs.io/zh_CN/latest/c12/p14_launching_daemon_process_on_unix.html"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.9.7 ('base')",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.7"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "caf1c2fcf97217de91eafa76b907d50f9ea378f5ffbee7f571142d119bb6a771"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
